package com.fivefu.core.skyeye.event.config;

import com.fivefu.core.skyeye.event.listener.MapRedisStreamListener;
import com.fivefu.core.skyeye.event.properties.RedisStringStreamBean;
import com.fivefu.core.skyeye.event.properties.StreamBean;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.ApplicationArguments;
import org.springframework.boot.ApplicationRunner;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.connection.stream.*;
import org.springframework.data.redis.serializer.StringRedisSerializer;
import org.springframework.data.redis.stream.StreamMessageListenerContainer;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.stereotype.Component;

import java.time.Duration;
import java.util.List;

@Component
public class MapRedisStreamRunner implements ApplicationRunner {

    private Logger logger = LoggerFactory.getLogger(MapRedisStreamRunner.class);

    @Autowired
    private RedisStringStreamBean stringStreamBean;

    @Autowired
    private RedisConnectionFactory redisConnectionFactory;
    @Autowired
    private ThreadPoolTaskExecutor threadPoolTaskExecutor;
    @Autowired
    private MapRedisStreamListener mapRedisStreamListener;

    private StreamMessageListenerContainer<String, MapRecord<String, String, String>> streamContainer;

    @Override
    public void run(ApplicationArguments args) throws Exception {
        if (stringStreamBean.getMapRedis() != null && !stringStreamBean.getMapRedis().isEmpty()){
            streamContainer = StreamMessageListenerContainer.create(this.redisConnectionFactory,options());
            //开始监听消息
            List<StreamBean> streamList = stringStreamBean.getMapRedis();
            streamList.forEach(stream -> {
                streamContainer.receive(Consumer.from(stream.getStreamGroup(), stream.getStreamConsumer()),
                        StreamOffset.create(stream.getStreamName(), ReadOffset.lastConsumed()),
                        this.mapRedisStreamListener);
            });
            //开始监听消息
            /*streamContainer.receive(Consumer.from(streamGroup, streamConsumer),
                    StreamOffset.create(streamName, ReadOffset.lastConsumed()),
                    this.redisStreamListener);*/
            this.streamContainer.start();
            logger.info("Redis监听启动成功");
        }

    }

    public StreamMessageListenerContainer.StreamMessageListenerContainerOptions<String, MapRecord<String, String, String>> options(){
        StreamMessageListenerContainer.StreamMessageListenerContainerOptions<String, MapRecord<String, String, String>> option = StreamMessageListenerContainer.StreamMessageListenerContainerOptions
                .builder()
                .batchSize(10)//一次性最多拉取数据数量
                .executor(this.threadPoolTaskExecutor)//执行消息轮询的执行器
                //消息消费异常的handler
                .errorHandler(t -> System.err.println("redis msg listener error，e:{}" + t))//超时时间，设置为0，表示不超时(超时后会抛出异常)
                .pollTimeout(Duration.ZERO)
                //序列化器
                .serializer(new StringRedisSerializer())
                .build();
        return option;
    }
}
